package com.java.springcloud.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author dushougudu
 */
@Slf4j
@Service
@RocketMQTransactionListener
public class RocketMqTransListener implements RocketMQLocalTransactionListener {

    private ConcurrentHashMap<String, Object> localTrans = new ConcurrentHashMap<>();

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            log.info("【本地业务执行完毕】 msg:{}, Object:{}", message, o);
            localTrans.put(message.getHeaders().getId() + "", message.getPayload());
            String payload = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);

            if (payload.equals("rollback")) {
                return RocketMQLocalTransactionState.ROLLBACK;
            } else if (payload.equals("unknown")) {
                return RocketMQLocalTransactionState.UNKNOWN;
            }

            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【执行本地业务异常】 exception message:{}", e.getMessage());
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("【执行检查任务】 msg:{}", msg);
        return RocketMQLocalTransactionState.COMMIT;
    }
}
